package com.hz.cheetah.client;

import com.hz.cheetah.common.Constants;
import com.hz.cheetah.common.bean.RpcRequest;
import com.hz.cheetah.common.bean.RpcResponse;
import com.hz.cheetah.common.codec.RpcDecoder;
import com.hz.cheetah.common.codec.RpcEncoder;
import com.hz.cheetah.hub.protocol.CheetahURL;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by yangchuan on 16/6/30.
 */
public class NettyClientProxy {

    private static final Logger LOGGER = LoggerFactory.getLogger(CheetahClient.class);


    private static final ConcurrentMap<String, NettyClientProxy> channelMap = new ConcurrentHashMap<>();

    private CheetahURL url;

    private Bootstrap bootstrap = null;

    private Channel channel;

    private EventLoopGroup group = null;

    private ChannelHandler handler = null;


    public NettyClientProxy(CheetahURL url,ChannelHandler handler){
        this.url = url;
        this.handler = handler;
    }

    public void doOpen(){
        bootstrap = new Bootstrap();
        group = new NioEventLoopGroup(Constants.DEFAULT_IO_THREADS);
        bootstrap.group(group);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Constants.DEFAULT_CONNECT_TIMEOUT);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new RpcEncoder(RpcRequest.class)); // 编码 RPC 请求
                pipeline.addLast(new RpcDecoder(RpcResponse.class)); // 解码 RPC 响应
                pipeline.addLast(handler); // 处理 RPC 响应
            }
        });
    }

     public void doConnection() throws Exception{
         long start = System.currentTimeMillis();
         ChannelFuture future = bootstrap.connect(url.getAdress(),url.getPort());
         try{
             boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);

             if (ret && future.isSuccess()) {
                 Channel newChannel = future.channel();

                 //newChannel.setInterestOps(Channel.OP_READ_WRITE);
                 try {
                     // 关闭旧的连接
                     Channel oldChannel = this.channel; // copy reference
                     if (oldChannel != null) {
                         try {
                             if (LOGGER.isInfoEnabled()) {
                                 LOGGER.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
                             }
                             oldChannel.close();
                         } finally {
                            // NettyChannel.removeChannelIfDisconnected(oldChannel);
                         }
                     }
                 } finally {
                     if (this.isClosed()) {
                         try {
                             if (LOGGER.isInfoEnabled()) {
                                 LOGGER.info("Close new netty channel " + newChannel + ", because the client closed.");
                             }
                             newChannel.close();
                         } finally {
                             this.channel = null;
                             //NettyChannel.removeChannelIfDisconnected(newChannel);
                         }
                     } else {
                         this.channel = newChannel;
                     }
                 }
             } else if (future.cause() != null) {
                 throw new RuntimeException( "client(url: " + url + ") failed to connect to server "
                         + url.getAdress() + ", error message is:" + future.cause().getMessage(), future.cause());
             } else {
                 throw new RuntimeException( "client(url: " + url + ") failed to connect to server "
                         + url.getAdress() + " client-side timeout "
                         + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client ");
             }
         }finally{
             if (! isConnected()) {
                 future.cancel(true);
             }
         }
     }

    private boolean isConnected(){
        return channel != null && channel.isActive();
    }

    private boolean isClosed(){
        return channel == null || !channel.isActive();
    }

    private long getConnectTimeout(){
        return Constants.DEFAULT_CONNECT_TIMEOUT;
    }

    public void close(){
        if(!isClosed()){
            channel.close();
        }
    }
}
